package com.ksyun.campus.client;

import com.ksyun.campus.client.domain.ApiResponse;
import com.ksyun.campus.client.domain.ReplicaData;

import com.ksyun.campus.client.domain.dto.FileBlockInfo;
import com.ksyun.campus.client.domain.dto.WriteContentDTO;

import com.ksyun.campus.client.domain.json.MetaServer;
import com.ksyun.campus.client.util.ZkUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.*;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.io.OutputStream;
import java.util.List;

@Data
@AllArgsConstructor
@NoArgsConstructor
@Slf4j
public class FSOutputStream extends OutputStream {

    private List<ReplicaData> dataServerList;

    private RestTemplate restTemplate=new RestTemplate();

    public FSOutputStream(List<ReplicaData> replicaDataList) {
        this.dataServerList=replicaDataList;
    }

    @Override
    public void write(int b) throws IOException {
        byte[] bytes = new byte[1];
        bytes[0] = (byte) b;
        write(bytes, 0, 1);
    }

    @Override
    public void write(byte[] b) throws IOException {
        write(b, 0, b.length);
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
        String path = "";
        for (ReplicaData replicaData :dataServerList) {
            String dsNode = replicaData.getDsNode();//ip+port
            path = replicaData.getPath();
            HttpHeaders headers=new HttpHeaders();

            WriteContentDTO writeContentDTO = new WriteContentDTO();
            writeContentDTO.setPath(path);
            writeContentDTO.setOffset(off);
            writeContentDTO.setLength(len);
            writeContentDTO.setBytes(b);
            HttpEntity<WriteContentDTO> request=new HttpEntity<>(writeContentDTO,headers);
            String dsURL="http://"+dsNode+"/write";
            ResponseEntity<ApiResponse> response = restTemplate.exchange(
                    dsURL,
                    HttpMethod.POST,
                    request,
                    ApiResponse.class);
            ApiResponse body = response.getBody();
            int code = body.getCode();
            if (code == 200) {
                log.info("数据写入分布式结点{}成功",dsNode);
            } else {
                throw new IOException("文件写入失败： " + body.getMessage());
            }
        }

        //更新元数据存储的文件大小

        MetaServer metaServer = ZkUtil.getMetaServer("master");
        if (metaServer==null){
            metaServer=ZkUtil.getMetaServer("slave");
        }
        String host = metaServer.getHost();
        Integer port = metaServer.getPort();
        String msURL="http://"+host+":"+port+"/write";

        FileBlockInfo fileBlockInfo=new FileBlockInfo();
        fileBlockInfo.setPath(path);
        fileBlockInfo.setLength(len);
        fileBlockInfo.setOffset(off);
        fileBlockInfo.setReplicas(dataServerList);

        HttpHeaders headers=new HttpHeaders();

        HttpEntity<FileBlockInfo> msRequest=new HttpEntity<>(fileBlockInfo,headers);

        ResponseEntity<ApiResponse> response = restTemplate.exchange(
                msURL,
                HttpMethod.POST,
                msRequest,
                ApiResponse.class);
        ApiResponse body = response.getBody();
        int code = body.getCode();
        if(code==200){
            log.info("元数据更新成功");
        }
    }

    @Override
    public void close() throws IOException {
        super.close();
    }

}
